package com.lsx143.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.lsx143.realtime.app.BaseApp;
import com.lsx143.realtime.common.Constants;
import com.lsx143.realtime.util.KafkaUtil;
import com.lsx143.realtime.util.MyUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.List;

/**
 * DWM层-UV-APP
 */
public class DWMUVApp extends BaseApp {
    public static void main(String[] args) {
        //消费dwd_page_log进行轻度统计uv指标
        new DWMUVApp().init(
                20001,
                "DWMUVApp",
                1,
                "DWMUVApp",
                Constants.TOPIC_DWD_PAGE_LOG);
    }

    /**
     * DWM-UV指标业务逻辑
     *
     * @param env       flink的执行环境
     * @param srcStream 源数据的流
     */
    @Override
    protected void run(StreamExecutionEnvironment env, DataStreamSource<String> srcStream) {
        System.out.println("【DWMUVApp】启动");
        srcStream
                //1.转换数据,String->Json
                .map(JSON::parseObject)
                //2.添加水印
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((ele, ts) -> ele.getLong("ts"))
                )
                //3.keyedBy,按照mid分组
                .keyBy(t -> t.getJSONObject("common").getString("mid"))
                //4.滚动窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))

                //5.process,对窗口数据进行逻辑处理
                .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {
                    //格式化日期,用于判断是否隔天登录
                    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                    //用于保存初次登录的时间
                    private ValueState<Long> firstLoginState;

                    @Override
                    public void open(Configuration parameters) {
                        firstLoginState = getRuntimeContext()
                                .getState(new ValueStateDescriptor<>("firstLoginState", Long.class));
                    }

                    @Override
                    public void process(String mid,
                                        Context ctx,
                                        Iterable<JSONObject> it,
                                        Collector<JSONObject> out) throws Exception {
                        //1.判断 是否初次登录 or 隔天登录
                        String now = sdf.format(new Date(ctx.currentWatermark()));
                        //这里隔天登录写在if判断中,可避免 firstLoginState=null 时的异常
                        if (firstLoginState.value() == null || !now.equals(sdf.format(firstLoginState.value()))) {
                            //转换为list处理
                            List<JSONObject> datas = MyUtil.toList(it);
                            //找到ts时间最小的数据,输出并保存到状态
                            JSONObject minData = Collections.min(datas, Comparator.comparing(o -> o.getLong("ts")));
                            out.collect(minData);
                            firstLoginState.update(minData.getLong("ts"));
                        }
                    }
                })
                //6.sink,写出数据到kafka
                .map(JSONAware::toJSONString)
                .addSink(KafkaUtil.getKafkaSink(Constants.TOPIC_DWM_UV));
    }
}
